package com.gitee.cnlongs.cnlongmq.server.threads.datasync;

import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.gitee.cnlongdb.basic.entity.datasync.CnLongDataSyncRequest;
import com.gitee.cnlongdb.basic.entity.datasync.CnLongDataSyncResponse;
import com.gitee.cnlongdb.basic.enums.sysnc.LongdbDataSyncEnms;
import com.gitee.cnlongdb.basic.utils.RandomUtil;
import com.gitee.cnlongdb.server.datasync.DataSyncConnectMessage;
import com.gitee.cnlongs.cnlongmq.basic.constant.CnLongMqConstant;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.request.CnLongMqConfiger;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqDataSync;
import com.gitee.cnlongs.cnlongmq.server.threads.LongMqDataSyncHandler;

/**
 * 系统名称：CnLongMqExchangeThread.java
 * 模块名称：
 * 模块描述：执行队列信息交换的操作
 * 功能列表：
 * 模块作者：ZOUYONG
 * 开发时间：2019年4月2日 下午10:05:55
 * 模块路径：com.gitee.cnlongs.cnlongmq.server.datasync.threads.CnLongMqExchangeThread
 * 更新记录：
 */
public class CnLongMqExchangeThread extends Thread {

	private static final Logger logger = (Logger) LoggerFactory.getLogger(CnLongMqExchangeThread.class);

	// 队列配置对象
	private CnLongMqConfiger longMqConfig;
	
	// 不需要通知的客户端列表
	private String filterCustomer;
	
	public CnLongMqExchangeThread(CnLongMqConfiger longMqConfig, String filterCustomer) {
		this.longMqConfig = longMqConfig;
		this.filterCustomer = filterCustomer;
	}
	
	@Override
	public void run() {
		try {
			List<String> customerList = longMqConfig.getCustomerList();
			if(CollectionUtils.isEmpty(customerList)) {
				logger.info("没有需要同步的队列数据！");
				return;
			}
			for (String customer : customerList) {
				if(StringUtils.isBlank(customer) || customer.equals(filterCustomer)) {
					continue;
				}
				String[] clientAry = customer.split(CnLongMqConstant.longmqSeparateF);
				try {
					CnLongDataSyncRequest dataSync = new CnLongDataSyncRequest();
					dataSync.setMessageType(CnLongMqDataSync.syncSeqList.name());
					dataSync.setClassPath(LongMqDataSyncHandler.class.getName());
					dataSync.setMethodName("startExchangeSeqList");
					dataSync.setParamClas(new String[]{ longMqConfig.getClass().getName() });
					dataSync.setParamObject(new String[]{ JSON.toJSONString(longMqConfig) });
					dataSync.setOperThreadId(RandomUtil.getCharAndNumr(20));
					
					// 开始发送netty消息，通知关联队列方进行队列更换
					CnLongDataSyncResponse response = 
							DataSyncConnectMessage.noticeDataSyncSend(clientAry[0], Integer.valueOf(clientAry[1]), dataSync);
					String jsonData = null != response ? response.getJsonData() : null;
					if(StringUtils.isNotBlank(jsonData) && 
							!LongdbDataSyncEnms.fail.name().equals(jsonData) && 
							!LongdbDataSyncEnms.error.name().equals(jsonData))
					{
						CnLongMqConfiger syncLongMqConfig = JSON.parseObject(jsonData, CnLongMqConfiger.class);
						LongMqDataSyncHandler.replayExchangeSeqList(syncLongMqConfig, false);
					}
				}
				catch(Exception e) {
					logger.error("队列同步异常：" + e.getMessage());
				}
			}
		}
		catch(Exception e) {
			logger.error("队列数据同步异常:" + e.getMessage());
		}
	}
	
}
